package com.sword.spring.rabbitmq.b_workqueue;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * com.sword.spring.rabbitmq.a_helloworld.Recv
 * Created by SwordLiu on 2017-11-02.
 */
public class Worker {

    private final static String QUEUE_NAME = "hello";
    private final static String DURABLE_QUEUE = "task_queue";

    /**
     * By default, RabbitMQ will send each message to the next consumer, in sequence.
     * On average every consumer will get the same number of messages. This way of distributing messages is called round-robin.
     * 默认情况下，RabbitMQ将按顺序将每条消息发送给下一个消费者。平均每个消费者将获得相同数量的消息。这种分发消息的方式叫做循环（round-robin）。
     * @throws IOException
     * @throws TimeoutException
     */
    public void receive() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(DURABLE_QUEUE, true, false, false, null);
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);

        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
                try {
                    doWork(message);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false; // acknowledgment is covered below
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private void doWork(String task) throws InterruptedException {
        for (char ch : task.toCharArray()) {
            if(ch == '.') {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }


    public static void main(String[] args) throws IOException, TimeoutException {
        new Worker().receive();
    }


}
